package com.shop.cloud.live.admin.socket;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.shop.cloud.common.core.util.R;
import com.shop.cloud.common.data.tenant.TenantContextHolder;
import com.shop.cloud.live.admin.service.LiveRoomService;
import com.shop.cloud.live.admin.service.LiveRoomSpuService;
import com.shop.cloud.live.admin.util.MapUtil;
import com.shop.cloud.live.common.constant.LiveConstant;
import com.shop.cloud.live.common.constant.LiveRoomStatusType;
import com.shop.cloud.live.common.constant.MsgCode;
import com.shop.cloud.live.common.constant.MsgType;
import com.shop.cloud.live.common.dto.LiveMessageDTO;
import com.shop.cloud.live.common.entity.LiveRoom;
import com.shop.cloud.live.common.entity.LiveRoomSpu;
import com.shop.cloud.live.common.vo.LiveMsgItemVO;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 用户ws
 * userType 0游客 1用户
 * @author 
 */
@ServerEndpoint("/ws/user/{tenantId}/{userType}/{userId}/{roomId}")
@Component
public class LiveUserWebSocketServer {

    public static final Logger logger = LoggerFactory.getLogger(LiveAnchorWebSocketServer.class);

    public static final ConcurrentHashMap<String, Session> LIVE_USER_MAP = new ConcurrentHashMap<>(LiveConstant.DEFAULT_MAP_SIZE);

    private static LiveRoomService liveRoomService;

    private static LiveRoomSpuService liveRoomSpuService;

    @Autowired
    public void setLiveRoomService(LiveRoomService liveRoomService) {
        LiveUserWebSocketServer.liveRoomService = liveRoomService;
    }

    @Autowired
    public void setLiveRoomSpuService(LiveRoomSpuService liveRoomSpuService) {
        LiveUserWebSocketServer.liveRoomSpuService = liveRoomSpuService;
    }

    @OnOpen
    public void onOpen(Session session, @PathParam("tenantId") String tenantId, @PathParam("roomId") String roomId,
                       @PathParam("userType") String userType, @PathParam("userId") String userId) throws IOException {
        TenantContextHolder.setTenantId(tenantId);
        logger.info("创建用户直播连接");
        // 检查直播间
        LiveRoom liveRoom = liveRoomService.getOne(Wrappers.<LiveRoom>lambdaQuery()
                .ne(LiveRoom::getLiveStatus, LiveRoomStatusType.OFFLINE.value())
                .eq(LiveRoom::getId, roomId)
                .gt(LiveRoom::getEndTime, LocalDateTime.now()));
        if (Objects.isNull(liveRoom)) {
            logger.error("直播间不存在");
            if (session.isOpen()) {
                // 当前未有直播权限
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CAN_NOT_FIND_ROOM_INFO, "直播间不存在")));
            }
            return;
        }
        Session oldSession = LIVE_USER_MAP.get(LiveConstant.LIVE_USER_PREFIX + liveRoom.getId() + "_" + userId);
        if (Objects.nonNull(oldSession)) {
            if (oldSession.isOpen()) {
                oldSession.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.LOGIN_IN_OTHER_PLACE, "账户已在别处登录，请刷新")));
                oldSession.close();
            }
            return;
        }
        // 存储会话到map
        LIVE_USER_MAP.put(LiveConstant.LIVE_USER_PREFIX + liveRoom.getId() + "_" + userId, session);
        if (session.isOpen()) {
            session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok()));
        }
        // 给所有用户更新直播观看人数
        sendChangeLiveUserCountMsg(roomId, LIVE_USER_MAP.size(), session);
        // 如果有商品是讲解状态，发送一条讲解消息给用户
        sendProdExplainingMsg(session, liveRoom);
    }

    private static void sendProdExplainingMsg(Session session, LiveRoom liveRoom) throws IOException {
        List<LiveRoomSpu> prodList = liveRoomSpuService.list(Wrappers.<LiveRoomSpu>lambdaQuery().eq(LiveRoomSpu::getRoomId, liveRoom.getId()));
        List<LiveRoomSpu> livingProds = prodList.stream().filter(p -> Objects.equals(p.getExplainStatus(), 1)).toList();
        if (CollUtil.isNotEmpty(livingProds)) {
            LiveRoomSpu liveRoomProdVO = livingProds.get(0);
            LiveMsgItemVO liveMsgItemVO = new LiveMsgItemVO();
            liveMsgItemVO.setMsgType(MsgType.EXPLAIN_STATUS);
            liveMsgItemVO.setSpuId(liveRoomProdVO.getSpuId());
            liveMsgItemVO.setExplainStatus("1");
            if (session.isOpen()) {
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(liveMsgItemVO)));
            }
        }
    }

    private static void sendChangeLiveUserCountMsg(String roomId, Integer liveUserCount, Session session) throws IOException {
        // 发送消息通知用户直播状态变更
        List<Session> onlineUserList = LiveUserWebSocketServer.getOnlineUserList(roomId);
        // 添加当前用户
        onlineUserList.add(session);
        LiveMsgItemVO liveMsgItem = new LiveMsgItemVO();
        liveMsgItem.setMsgType(MsgType.LIVE_USER_COUNT);
        liveMsgItem.setLiveUserCount(liveUserCount);
        if (CollUtil.isNotEmpty(onlineUserList)) {
            for (Session onlineUser : onlineUserList) {
                if (Objects.nonNull(onlineUser) && onlineUser.isOpen()) {
                    onlineUser.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(liveMsgItem)));
                }
            }
        }
    }

    @OnClose
    public void onClose(Session session, @PathParam("tenantId") String tenantId,
                        @PathParam(value = "userId") Long userId, @PathParam(value = "roomId") String roomId) throws IOException {
        TenantContextHolder.setTenantId(tenantId);
        logger.info("关闭用户直播连接");
        LIVE_USER_MAP.remove(LiveConstant.LIVE_USER_PREFIX + roomId + "_" + userId);
        // 给所有用户更新直播观看人数
        sendChangeLiveUserCountMsg(roomId, LIVE_USER_MAP.size(), session);
    }

    @OnError
    public void onError(Session session, Throwable error, @PathParam("tenantId") String tenantId,
                        @PathParam(value = "userId") String userId, @PathParam("roomId") String roomId) throws IOException {
        TenantContextHolder.setTenantId(tenantId);
        logger.error("用户直播连接发生错误{}", error.toString());
        LIVE_USER_MAP.remove(LiveConstant.LIVE_USER_PREFIX + roomId + "_" + userId);
        // 给所有用户更新直播观看人数
        sendChangeLiveUserCountMsg(roomId, LIVE_USER_MAP.size(), session);
    }

    @OnMessage
    public void onMessage(String sourceMessage, Session session, @PathParam("tenantId") String tenantId,
                          @PathParam("roomId") String roomId, @PathParam("userType") Integer userType, @PathParam("userId") String userId) throws IOException {
        TenantContextHolder.setTenantId(tenantId);
        if (StrUtil.isBlank(sourceMessage)) {
            if (session.isOpen()) {
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CONTENT_FORMAT_ERROR, "直播用户消息内容格式有误")));
            }
            return;
        }
        logger.info("直播用户sourceMessage:" + sourceMessage);
        LiveMessageDTO liveMessageInfo = JSONUtil.toBean(sourceMessage, LiveMessageDTO.class);
        if (Objects.isNull(liveMessageInfo)) {
            if (session.isOpen()) {
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CONTENT_FORMAT_ERROR, "直播用户消息内容格式有误")));
            }
            return;
        }
        if (checkLiveRoom(session, roomId)) {
            return;
        }
        // 保持心跳
        if (Objects.equals(liveMessageInfo.getMsgType(), MsgType.HEART_BEAT)) {
            if (session.isOpen()) {
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(MsgCode.HEART_BEAT, "保持心跳")));
            }
            return;
        }
        if (Objects.equals(userType, 0) && Objects.equals(liveMessageInfo.getMsgType(), MsgType.BULLET_SCREEN)) {
            // 游客 不能发送弹幕
            return;
        }
        // 用户发送弹幕
        if (Objects.equals(liveMessageInfo.getMsgType(), MsgType.BULLET_SCREEN)) {
            if (StrUtil.isBlank(liveMessageInfo.getContent())) {
                return;
            }
            // 弹幕内容不超过50个字符
            if (liveMessageInfo.getContent().length() > LiveConstant.BULLET_SCREEN_MAX_LENGTH) {
                if (session.isOpen()) {
                    session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CONTENT_FORMAT_ERROR, "用户弹幕内容过长")));
                }
                return;
            }
            // 发送弹幕给在线用户
            sendMsgToUser(roomId, liveMessageInfo.getNickName(), liveMessageInfo.getContent(), "0");
        }
    }

    private static boolean checkLiveRoom(Session session, String roomId) throws IOException {
        // 检查直播间
        LiveRoom liveRoom = liveRoomService.getOne(Wrappers.<LiveRoom>lambdaQuery()
                .ne(LiveRoom::getLiveStatus, LiveRoomStatusType.OFFLINE.value())
                .eq(LiveRoom::getId, roomId)
                .gt(LiveRoom::getEndTime, LocalDateTime.now()));
        if (Objects.isNull(liveRoom)) {
            logger.error("直播间不存在");
            if (session.isOpen()) {
                // 直播间不存在
                session.getBasicRemote().sendText(JSONUtil.toJsonStr(R.failed(MsgCode.CAN_NOT_FIND_ROOM_INFO, "直播间不存在")));
            }
            return true;
        }
        return false;
    }

    public static void sendMsgToUser(String roomId, String nickName, String content, String userType) throws IOException {
        // 给在线用户发送弹幕消息
        List<Session> onlineUserList = getOnlineUserList(roomId);
        if (CollUtil.isNotEmpty(onlineUserList)) {
            LiveMsgItemVO liveMsgItem = new LiveMsgItemVO();
            liveMsgItem.setMsgType(MsgType.BULLET_SCREEN);
            liveMsgItem.setNickName(nickName);
            liveMsgItem.setContent(content);
            liveMsgItem.setUserType(userType);
            for (Session onlineUser : onlineUserList) {
                if (Objects.nonNull(onlineUser) && onlineUser.isOpen()) {
                    onlineUser.getBasicRemote().sendText(JSONUtil.toJsonStr(R.ok(liveMsgItem)));
                }
            }
        }
    }

    public static List<Session> getOnlineUserList(String roomId) {
        // 查找当前直播间在线用户
        List<Session> onlineUserList = new ArrayList<>(LiveConstant.DEFAULT_MAP_SIZE);
        for (String liveUserKey : LiveUserWebSocketServer.LIVE_USER_MAP.keySet()) {
            // 给在线用户发送消息
            if (liveUserKey.startsWith(LiveConstant.LIVE_USER_PREFIX + roomId + "_")) {
                onlineUserList.add(LiveUserWebSocketServer.LIVE_USER_MAP.get(liveUserKey));
            }
        }
        // 给主播发送消息
        LiveRoom liveRoom = liveRoomService.getOne(Wrappers.<LiveRoom>lambdaQuery()
                .ne(LiveRoom::getLiveStatus, LiveRoomStatusType.OFFLINE.value())
                .eq(LiveRoom::getId, roomId)
                .gt(LiveRoom::getEndTime, LocalDateTime.now()));
        if (Objects.isNull(liveRoom)) {
            List<Session> sessions = MapUtil.getSessionByKeyPrefix(LiveAnchorWebSocketServer.LIVE_ANCHOR_MAP, LiveConstant.LIVE_ANCHOR_PREFIX + roomId + "_");
            if (CollUtil.isNotEmpty(sessions)) {
                onlineUserList.addAll(sessions);
            }
        } else {
            Session anchorSession = LiveAnchorWebSocketServer.LIVE_ANCHOR_MAP.get(LiveConstant.LIVE_ANCHOR_PREFIX + roomId + "_" + liveRoom.getUserId());
            if (Objects.nonNull(anchorSession)) {
                onlineUserList.add(anchorSession);
            }
        }
        return onlineUserList;
    }
}
